package org.wu.framework.lazy.orm.database.sql.expand.database.persistence.event;

import org.framework.wu.framework.queue.Message;
import org.framework.wu.framework.queue.MessageQueue;
import org.framework.wu.framework.queue.MessageQueueFactory;
import org.wu.framework.core.CamelAndUnderLineConverter;
import org.wu.framework.core.utils.ObjectUtils;
import org.wu.framework.lazy.orm.database.lambda.stream.LambdaTableType;
import org.wu.framework.lazy.orm.database.lambda.stream.lambda.LazyLambdaStream;
import org.wu.framework.lazy.orm.database.sql.expand.database.persistence.method.LazyOperationParameter;
import org.wu.framework.lazy.orm.database.sql.util.SqlUtils;
import org.wu.framework.translation.data.convert.LazyDataFactory;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class LazyOrmEventConsumer {

    private final LazyOperationParameter lazyOperationParameter;

    private final List<LazyOrmEvent<?>> lazyOrmEvents;
    private final LazyLambdaStream lazyLambdaStream;

    ThreadPoolExecutor LAZY_ORM_EVENT_EXECUTOR =
            new ThreadPoolExecutor(10, 20, 3L, TimeUnit.MINUTES,
                    new LinkedBlockingDeque<>(50), new ThreadPoolExecutor.AbortPolicy());


    public LazyOrmEventConsumer(LazyOperationParameter lazyOperationParameter, List<LazyOrmEvent<?>> lazyOrmEvents, LazyLambdaStream lazyLambdaStream) {
        this.lazyOperationParameter = lazyOperationParameter;
        this.lazyOrmEvents = lazyOrmEvents;
        this.lazyLambdaStream = lazyLambdaStream;
    }


    public void run() {
        String eventQueueName = lazyOperationParameter.getLazyOperationAttribute().getEventQueueName();
        MessageQueue queue = MessageQueueFactory.getQueue(eventQueueName);
        Map<String, ? extends LazyOrmEvent<?>> LazyOrmEvent_Cache_Map =
                lazyOrmEvents
                        .stream()
                        .collect(
                                Collectors.toMap(LazyOrmEvent::getEventTable, lazyOrmEvent -> lazyOrmEvent, (A, B) -> A)
                        );
        // 创建监听线程
        Thread thread = new Thread(() -> {
            while (true) {
                Message receive = queue.receive();
                if (ObjectUtils.isNotEmpty(receive)) {
                    Object sql = receive.getBody();
                    String topic = receive.getTopic();
                    List<String> tablesInSql = SqlUtils.tablesInSql(sql.toString());
                    for (String tableName : tablesInSql) {

                        try {
                            if (LazyOrmEvent_Cache_Map.containsKey(tableName)) {
                                LazyOrmEvent<?> lazyOrmEvent = LazyOrmEvent_Cache_Map.get(tableName);
                                Class<?> payloadDataClass = lazyOrmEvent.payloadDataClass();
                                if (LambdaTableType.INSERT.name().equals(topic)) {
                                    List<Map<String, Object>> dataList = SqlUtils.dataInSql(sql.toString());
                                    for (Map<String, Object> data : dataList) {
                                        // 修改数据库字段带有下划线，Java实体驼峰问题
                                        Map<String, Object> javaBean = new HashMap<>();
                                        data.forEach((key, value) -> {
                                            String javaKey= CamelAndUnderLineConverter.lineToHumpField(key);
                                            javaBean.put(javaKey, value);
                                        });
                                        Object entity = LazyDataFactory.handler(javaBean, payloadDataClass);
                                        Method declaredMethod = lazyOrmEvent.getClass().getDeclaredMethod("insertEvent", payloadDataClass);
                                        declaredMethod.invoke(lazyOrmEvent, entity);
                                    }
                                } else if (LambdaTableType.UPDATE.name().equals(topic)) {
                                    // 通过where 查询数据
                                    String whereSql = SqlUtils.whereConditionsInSql(sql.toString());
                                    List<?> updateDataList = lazyLambdaStream.executeSQL("select * from {0} {1}", payloadDataClass,
                                            tableName, whereSql);
                                    for (Object updateData : updateDataList) {
                                        Method declaredMethod = lazyOrmEvent.getClass().getDeclaredMethod("updateEvent", payloadDataClass);
                                        declaredMethod.invoke(lazyOrmEvent, updateData);
                                    }

                                } else if (LambdaTableType.DELETE.name().equals(topic)) {
                                    // 通过where 查询数据
                                    String whereSql = SqlUtils.whereConditionsInSql(sql.toString());
                                    List<?> deleteDataList = lazyLambdaStream.executeSQL("select * from {0} {1}", payloadDataClass,
                                            tableName, whereSql);
                                    for (Object deleteData : deleteDataList) {
                                        Method declaredMethod = lazyOrmEvent.getClass().getDeclaredMethod("deleteEvent", payloadDataClass);
                                        declaredMethod.invoke(lazyOrmEvent, deleteData);
                                    }

                                }
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }


                    // 解析数据类型
                    // 解析数据
                }
            }

        });
        LAZY_ORM_EVENT_EXECUTOR.submit(thread::start);
    }
}
